package com.lanyou.cook.camel.service;

import java.util.concurrent.TimeUnit;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.log4j.Logger;

import com.lanyou.cook.camel.commons.Constants;
import com.lanyou.cook.camel.util.ExceptionToStrFactory;
import com.lanyou.cook.camel.util.PropFactory;


/**
 * 与ZooKeeper通讯的基础设施
 * @author ly-lihailang
 */

public class CuratorService {
	private static Logger logger = Logger.getLogger(CuratorService.class);
	private CuratorFramework client;
	private ConnectionState lastConnectionState = ConnectionState.LOST;
	
	private String zkConnectString;
	
	/**
	 * 初始化创建CuratorFramework，并连接zookeeper服务
	 */
	public void init(){
		zkConnectString = PropFactory.getPropByKey(Constants.ZOOKEEPER_SERVER_URL);
		logger.info("[CuratorService.init]:zkConnectString="+zkConnectString);
		client = CuratorFrameworkFactory.builder()  
		        .connectString(zkConnectString).namespace(Constants.ZOOKEPPER_NAMESPACE) //连接zookeeper的ip、namespace 
		        .retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))  //重试指定的次数, 且每一次重试之间停顿的时间逐渐增加.
		        .connectionTimeoutMs(5000).build();//设置超时
		
		// 监听连接状态
//		client.getConnectionStateListenable().addListener((client2, newState) -> {
//			lastConnectionState = newState;
//		});
		//增加连接监听，判断CuratorFramework的连接状态
		client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
			
			@Override
			public void stateChanged(CuratorFramework arg0, ConnectionState newState) {
				lastConnectionState = newState;
			}
		});
		//启动CuratorFramework
		client.start();
		
		// 启动时连接zookeeper失败，则整个程序启动失败
		try {
			if(!client.blockUntilConnected(5, TimeUnit.SECONDS)){
				throw new Error("无法连接到ZooKeeper!");
			}
		} catch (InterruptedException e1) {
			logger.error("[CuratorService.init]:"+ExceptionToStrFactory.getErrorInfoFromException(e1));
			Thread.currentThread().interrupt();
		}
	}
	
	/**
	 * 实例注销时关闭CuratorFramework
	 */
	public void close(){
		closeClient();
	}
	
	/**
	 * 关闭CuratorFramework实例
	 */
	private void closeClient(){
		if(client != null){
			client.close();
			client = null;
		}
	}
	
	/**
	 * @return CuratorFramework实例
	 */
	public CuratorFramework getClient(){
		return client;
	}
	
	/**
	 * @return 连接状态
	 */
	public boolean isConnected(){
		return lastConnectionState.isConnected();
	}

	public String getZkConnectString() {
		return zkConnectString;
	}

	public void setZkConnectString(String zkConnectString) {
		this.zkConnectString = zkConnectString;
	}
	public static void main(String[] args) {
		new CuratorService().init();
	}
}
